package com.xiashitech.apiservice.grpc;

import cn.hutool.json.JSONUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.xiashitech.apiservice.Constants.EnAndDecryption;
import com.xiashitech.apiservice.Constants.ErrorCode;
import com.xiashitech.apiservice.config.AuthConfig;
import com.xiashitech.apiservice.dto.AuthFactDTO;
import com.xiashitech.apiservice.service.IAgentConfigService;
import com.xiashitech.apiservice.service.IAuthFactService;
import com.xiashitech.apiservice.service.IEnAndDecryption;
import com.xiashitech.apiservice.util.AuthFactSignParam;
import com.xiashitech.apiservice.util.EnAndDecryptionHolder;
import com.xiashitech.interfaces.agent.dto.AgentConfigDTO;
import com.xiashitech.interfaces.apiservice.dto.AuthFactRequest;
import com.xiashitech.interfaces.apiservice.dto.AuthFactResponse;
import com.xiashitech.interfaces.grpc.*;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Optional;
import java.util.concurrent.*;


@Slf4j
//@Service
@GrpcService
public class ConfigService extends RPCConfigServiceGrpc.RPCConfigServiceImplBase {
    @Autowired
    private IAgentConfigService agentConfigService;

    @Autowired
    private AuthConfig authConfig;

    @Autowired
    private EnAndDecryptionHolder enAndDecryptionHolder;
    @Autowired
    private IAuthFactService authFactService;

    private volatile boolean isHeartBeat = true;
    /**
     * 记录 responseObserver，用于给客户端推送
     */
    private final static HashSet<StreamObserver<RPCConfigResponse>> configResponseSet = new HashSet<>();

    private final ThreadPoolExecutor pushPool = new ThreadPoolExecutor(1,1,
            1, TimeUnit.SECONDS,
            new LinkedBlockingDeque<Runnable>(1),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.DiscardOldestPolicy() // 丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
            );
    private final Runnable pushRunnable;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    public ConfigService() {
        pushRunnable = new Runnable() {
            @Override
            public void run() {
                try {
                    log.info("pushConfigToAgent: hashset size: " + configResponseSet.size());
                    if(configResponseSet.size() <= 0)
                        return;
                    String configStr = "CONFIGHEARTBEAT"; // 传固定字符串，减少查库的压力
                    if(!isHeartBeat) {
                        isHeartBeat = true;
                        TimeUnit.MILLISECONDS.sleep(1500); // 等待 1.5s 后再执行推送任务，为了 nacos 中的配置信息已经更新下来
                        AgentConfigDTO config = agentConfigService.syncConfig();
                        configStr = JSONUtil.toJsonStr(config);
                        if(configStr == null|| "".equals(configStr)) {
                            log.info("pushConfigToAgent, the config string is empty");
                            return;
                        }
                    }
                    RPCConfigResponse response = RPCConfigResponse.newBuilder().setAgentConfigDTOJson(configStr).build();
                    int idx = 0;
                    synchronized (configResponseSet) { // configResponseSet 非线程安全，不同线程中增、删可能有问题
                        Iterator<StreamObserver<RPCConfigResponse>> iterator = configResponseSet.iterator();
                        while (iterator.hasNext()) {
                            idx++;
                            StreamObserver<RPCConfigResponse> soResponse = iterator.next();
                            try {
                                log.info("pushConfigToAgent: push start,  " + idx + ", " + soResponse + ", " + System.identityHashCode(soResponse));
                                ServerCallStreamObserver<RPCConfigResponse> scso = (ServerCallStreamObserver) soResponse;
                                if (!scso.isCancelled()) {
                                    soResponse.onNext(response); // 只会有一个线程执行，没有其他地方会调用对应对象的 onNext，无需加锁
                                    log.info("pushConfigToAgent: push end,  contentLength: " + configStr.length() + ", " + idx + ", " + soResponse + ", " + System.identityHashCode(soResponse));
                                } else {
                                    iterator.remove();
                                    log.info("pushConfigToAgent: push removed, channel canceled  " + idx + ", " + soResponse + ", " + System.identityHashCode(soResponse));
                                }
                            } catch (Exception e) {
                                // 认为推送失败，将 StreamObserver<RPCConfigResponse> 清除
                                log.info("pushConfigToAgent: push fail,  remove " + idx + ", " + soResponse + ", " + System.identityHashCode(soResponse));
                                iterator.remove();
                                e.printStackTrace();
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };

        // 当使用 agent <--> agentserver 之间代理时，需要定时推送，避免通道被代理停掉，注意与 nginx 的 grpc_read_timeout 180s 配置配套，要小于 180
        // 原先自己配置的 nginx 时，按以上要求，设置的 150，使用通用代理时，安全起见，改为 50s（通常 ngix 的设置是 60s）
        scheduledExecutorService.scheduleWithFixedDelay(pushRunnable, 30, 50, TimeUnit.SECONDS);
    }
    /**
     * 服务端推送，
     * 可阻塞和非阻塞方式访问
     * 阻塞方式的访问应答不缓存，只缓存需要推送的缓存
     * @param request
     * @param responseObserver
     */
    @Override
    public void grpcSyncConfig(RPCConfigRequest request, StreamObserver<RPCConfigResponse> responseObserver) {
        String sysInfo = request.getSysinfoJson();
        log.info("grpcSyncConfig channel build start, request json string: " + request.getSysinfoJson());

        String requestType = request.getType();
        AgentConfigDTO config = agentConfigService.syncConfig();
        String configStr = JSONUtil.toJsonStr(config);
        log.info("grpcSyncConfig config json string: " + configStr);
        RPCConfigResponse response = RPCConfigResponse.newBuilder().setAgentConfigDTOJson(configStr).build();
        responseObserver.onNext(response); // 此时的 responseObserver 只会单线程访问，不用加锁

        // 各个agent 只调用一次，如果调用多次，应该会报错，会出现推送多次的问题
        log.info("grpcSyncConfig  responseObserver: " + responseObserver + ", " + System.identityHashCode(responseObserver));
        synchronized (configResponseSet) {
            if ("nonblocking".equals(requestType) && !configResponseSet.contains(responseObserver)) {
                ServerCallStreamObserver scso = (ServerCallStreamObserver) responseObserver;
                scso.setOnCancelHandler(() -> {
                    // 客户端已经取消，服务端不用再缓存该 responseObserver
                    log.info("OnCancelHandler be called");
                    synchronized (configResponseSet) { // 这里应该是会并发的
                        configResponseSet.remove(scso);
                    }
                });
                log.info("grpcSyncConfig nonblocking, build a endpoint: " + sysInfo + ", add to hashset: " + responseObserver + ", " + System.identityHashCode(responseObserver));
                configResponseSet.add(responseObserver);
            } else if("blocking".equalsIgnoreCase(requestType)) {
                responseObserver.onCompleted();
                log.info("grpcSyncConfig  blocking return: " + responseObserver + ", " + System.identityHashCode(responseObserver));
            }
        }
        log.info("grpcSyncConfig response end");
    }

    /**
     * 为避免频繁的多次推送，使用单元素队列的线程池，丢弃的替换规则：新的替换旧的
     * @param
     */
    public void pushConfigToAgent() {
        isHeartBeat = false;
        pushPool.execute(pushRunnable);
    }

    @Override
    public void grpcInitSystemInfo(RPCSystemInfoRequest request, StreamObserver<RPCSystemInfoResponse> responseObserver) {
        String sysInfoStr = request.getSystemInfoRequestJson();
        AuthFactRequest authFactRequest = JSONUtil.toBean(sysInfoStr, AuthFactRequest.class);

        if(authConfig.getIsSign()) {
            String signType = Optional.ofNullable(authConfig.getSignType()).orElse(EnAndDecryption.SIGNTYPE_MD5);
            IEnAndDecryption enAndDecryption = enAndDecryptionHolder.getEnAndDecryptionMap().get(signType);
            Boolean verifySign = enAndDecryption.verifySign(authFactRequest.getSign(), AuthFactSignParam.generateSignParam(authFactRequest) + authConfig.getSecurityKey());
            if(!verifySign) {
                AuthFactResponse authFactResponse = AuthFactResponse.builder().code(ErrorCode.RESULT_4002.getCode()).message(ErrorCode.RESULT_4002.getMessage()).build();
                RPCSystemInfoResponse response = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(authFactResponse)).build();
                responseObserver.onNext(response);
                responseObserver.onCompleted();
                return;
            }
        }
        AuthFactDTO authFactDTO = authFactService.getSystemInfo(AuthFactDTO.builder().systemId(authFactRequest.getSystemId()).appCode(authFactRequest.getAppCode()).build());
        AuthFactResponse authFactResponse = AuthFactResponse.builder().systemInfoDTO(authFactDTO.getSystemInfoDTO()).code(authFactDTO.getCode()).message(authFactDTO.getMessage()).build();
        if(authFactResponse.getCode() == null) {
            authFactResponse.setCode(ErrorCode.RESULT_1001.getCode());
            authFactResponse.setMessage(ErrorCode.RESULT_1001.getMessage());
        } else {
            log.error("请求处理错误{}",authFactResponse.getMessage());
        }
        RPCSystemInfoResponse response = RPCSystemInfoResponse.newBuilder().setSystemInfoResponseJson(JSONUtil.toJsonStr(authFactResponse)).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}
